package com.gy.hadoop.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.Arrays;

/**
 * create 't_user_info_bak','base_info'
 */
public class HbaseMapperReduce {

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "centos102:2181,centos102:2181,centos103:2181");

        conf.set("fs.defaultFS", "hdfs://centos102:8020");
        conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resourcemanager.hostname", "http://centos103");
//        conf.set("yarn.nodemanager.aux-services", "mapreduce_shuffle");

        conf.set("mapreduce.job.jar", "/Users/mac/Downloads/workspace/kkb-bigdata/target/kkb-bigdata-1.0-SNAPSHOT.jar");
        conf.set("mapreduce.job.name", HbaseMapperReduce.class.getSimpleName());

        int runCode = ToolRunner.run(conf, new HbaseMR(), args);
        System.exit(runCode);
    }

    static class HbaseMR extends Configured implements Tool {

        @Override
        public int run(String[] args) throws Exception {
            Job job = Job.getInstance(super.getConf());
            job.setJarByClass(HbaseMapperReduce.class);

            TableMapReduceUtil.initTableMapperJob(
                    TableName.valueOf("t_user_info"),
                    new Scan(),
                    HbaseMapper.class,
                    Text.class,
                    Put.class,
                    job);

            TableMapReduceUtil.initTableReducerJob(
                    "t_user_info_bak",
                    HBaseReducer.class,
                    job);

            return job.waitForCompletion(true) ? 0 : 1;
        }
    }


    static class HbaseMapper extends TableMapper<Text, Put> {

        private final Text mapKey = new Text();

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            mapKey.set(Bytes.toString(key.get()));
            final Put mapVal = new Put(key.get());

            Arrays.stream(value.rawCells())
                    .forEach(x -> {
                        String family = Bytes.toString(CellUtil.cloneFamily(x));
                        if ("base_info".equals(family)) {
                            String qualifier = Bytes.toString(CellUtil.cloneQualifier(x));
                            if ("married".equals(qualifier) || "username".equals(qualifier)) {
                                try {
                                    mapVal.add(x);
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }

                        }

                    });

            if (!mapVal.isEmpty()) {
                context.write(mapKey, mapVal);
            }
        }
    }


    static class HBaseReducer extends TableReducer<Text, Put, ImmutableBytesWritable> {


        private final ImmutableBytesWritable rKey = new ImmutableBytesWritable();

        @Override
        protected void reduce(Text key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
            rKey.set(key.toString().getBytes());

            for (Put put : values) {
                context.write(rKey, put);
            }
        }
    }

}
